package study.mq.kafka.consumer.support.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import javax.annotation.PostConstruct;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
//@Configuration
//@AutoConfigureBefore(KafkaAutoConfiguration.class)
public class _2_KafkaListenerContainerConfig {

    private static final int CORE_SIZE = 2;

    @Autowired
    private ConcurrentKafkaListenerContainerFactory factory;
    @Autowired
    private ThreadPoolTaskExecutor taskExecutor;

    @PostConstruct
    public void init() {
        //容器属性
        ContainerProperties props = factory.getContainerProperties();
        //任务线程池
        props.setConsumerTaskExecutor(taskExecutor);
        //分区再平衡监听
//        props.setConsumerRebalanceListener(new CustomRebalanceListener());
    }

    @Bean
    public ThreadPoolTaskExecutor consumerTaskExecutor() {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();
        //
        taskExecutor.setCorePoolSize(CORE_SIZE);
        taskExecutor.setMaxPoolSize(CORE_SIZE);
        taskExecutor.setQueueCapacity(100);
        //线程名
        taskExecutor.setThreadNamePrefix("consumer-pool-thread-");
        //
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        taskExecutor.setWaitForTasksToCompleteOnShutdown(true);
        taskExecutor.initialize();
        return taskExecutor;
    }
}
